Akka Actor的创建&引用&声明周期

Posted on | In |

Actor的创建&引用&声明周期

1.创建actor

  • 定义一个Actor类

要定义自己的Actor类，需要继承Actor并实现receive方法。receive方法需要定义一系列case语句（类型为PartialFunction[Any, Unit]）来描述你的Actor能够处理哪些消息（使用标准的Scala模式匹配），以及消息如何被处理。
如下例：

1
2
3
4
5
6
7
8
9
10
11
import akka.actor.Actor
import akka.actor.Props
import akka.event.Logging

class MyActor extends Actor {
  val log = Logging(context.system, this)
  def receive = {
    case "test" => log.info("received test")
    case _      => log.info("received unknown message")
  }
}

  • Props

Props是一个用来在创建actor时指定选项的配置类，可以把它看作是不可变的，因此在创建包含相关部署信息的actor时（例如使用哪一个调度器(dispatcher)，详见下文），是可以自由共享的。以下是如何创建Props实例的示例.

1
2
3
4
5
import akka.actor.Props

val props1 = Props[MyActor]
val props2 = Props(new ActorWithArgs("arg")) // careful, see below
val props3 = Props(classOf[ActorWithArgs], "arg")

警告
在另一个actor中声明一个actor是非常危险的，会打破actor的封装。永远不要将一个actor的this引用传进Props！
推荐做法

在每一个Actor的伴生对象中提供工厂方法是一个好主意，这有助于保持创建合适的Props，尽可能接近actor的定义。这也避免了使用Props.apply(…)方法将采用一个“按名”（by-name）参数的缺陷，因为伴生对象的给定代码块中将不会保留包含作用域的引用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
object DemoActor {
  /**
   * Create Props for an actor of this type.
   * @param magciNumber The magic number to be passed to this actor’s constructor.
   * @return a Props for creating this actor, which can then be further configured
   *         (e.g. calling `.withDispatcher()` on it)
   */
  def props(magicNumber: Int): Props = Props(new DemoActor(magicNumber))
}

class DemoActor(magicNumber: Int) extends Actor {
  def receive = {
    case x: Int => sender() ! (x + magicNumber)
  }
}

class SomeOtherActor extends Actor {
  // Props(new DemoActor(42)) would not be safe
  context.actorOf(DemoActor.props(42), "demo")
  // ...
}

  • 使用Props创建Actor

Actor可以通过将Props实例传入actorOf工厂方法来创建，ActorSystem和ActorContext中都有该方法。

1
2
3
4
5
6
7
8
9
10
11
import akka.actor.ActorSystem

// ActorSystem is a heavy object: create only one per application
val system = ActorSystem("mySystem")
val myActor = system.actorOf(Props[MyActor], "myactor2")
使用ActorSystem将创建顶级actor，由actor系统提供的守护actor监管；如果使用的是actor的上下文，则创建一个该actor的子actor。

class FirstActor extends Actor {
  val child = context.actorOf(Props[MyActor], name = "myChild")
  // plus some behavior ...
}

推荐创建一个树形结构，包含子actor、孙子等等，使之符合应用的逻辑错误处理结构

  • 依赖注入

如果你的actor有带参数的构造函数，则这些参数也需要成为Props的一部分，如上文所述。但有些情况下必须使用工厂方法，例如，当实际构造函数的参数由依赖注入框架决定。

1
2
3
4
5
6
7
8
9
10
11
12
13
import akka.actor.IndirectActorProducer

class DependencyInjector(applicationContext: AnyRef, beanName: String)
  extends IndirectActorProducer {

  override def actorClass = classOf[Actor]
  override def produce =
    // obtain fresh Actor instance from DI framework ...
}

val actorRef = system.actorOf(
  Props(classOf[DependencyInjector], applicationContext, "hello"),
  "helloBean")

2.Actor API

Actor trait只定义了一个抽象方法，就是上面提到的receive，用来实现actor的行为。

如果当前actor的行为与收到的消息不匹配，则会调用 unhandled，其缺省实现是向actor系统的事件流中发布一条akka.actor.UnhandledMessage(message, sender, recipient)（将配置项akka.actor.debug.unhandled设置为on来将它们转换为实际的调试消息）。

另外，它还包括:

  • self引用代表本actor的ActorRef
  • sender引用代表最近收到消息的发送actor，通常用于下面将讲到的消息回应中
  • supervisorStrategy 用户可重写它来定义对子actor的监管策略

该策略通常在actor内声明，这样决定函数就可以访问actor的内部状态：因为失败通知作为消息发送给监管者，并像普通消息一样被处理（尽管不是正常行为），所有的值和actor变量都是可用的，以及sender引用 （报告失败的将是直接子actor；如果原始失败发生在遥远的后裔，它仍然是一次向上报告一层）。

  • context暴露actor和当前消息的上下文信息，如：
    • 用于创建子actor的工厂方法（actorOf）
    • actor所属的系统
    • 父监管者
    • 所监管的子actor
    • 生命周期监控
    • hotswap行为栈，见Become/Unbecome

3.Actor生命周期

enter image description here

actor系统中的路径代表一个”地方”，这里可能会被活着的actor占据。最初（除了系统初始化actor）路径都是空的。在调用actorOf()时它将为指定路径分配根据传入Props创建的一个actor化身。actor化身是由路径和一个UID标识的。重新启动只会替换有Props定义的Actor实例，但不会替换化身，因此UID保持不变。

当actor停止时，其化身的生命周期结束。在这一时间点上相关的生命周期事件被调用，监视该actor的actor都会获得终止通知。当化身停止后，路径可以重复使用，通过actorOf()创建一个actor。在这种情况下，除了UID不同外，新化身与老化身是相同的。

ActorRef始终表示化身（路径和UID）而不只是一个给定的路径。因此如果actor停止，并且创建一个新的具有相同名称的actor，则指向老化身的ActorRef将不会指向新的化身。

相对地，ActorSelection指向路径（或多个路径，如果使用了通配符），且完全不关注有没有化身占据它。因此ActorSelection 不能被监视。获取某路径下的当前化身ActorRef是可能的，只要向该ActorSelection发送Identify，如果收到ActorIdentity回应，则正确的引用就包含其中（详见通过Actor Selection确定Actor）。也可以使用ActorSelection的resolveOne方法，它会返回一个包含匹配ActorRef的Future。

  • 使用DeathWatch进行生命周期监控

为了在其它actor终止时 (即永久停止，而不是临时的失败和重启)收到通知，actor可以将自己注册为其它actor在终止时所发布的Terminated消息的接收者（见停止 Actor）。这个服务是由actor系统的DeathWatch组件提供的。

注册一个监视器很简单：

1
2
3
4
5
6
7
8
9
10
11
12
13
import akka.actor.{ Actor, Props, Terminated }

class WatchActor extends Actor {
  val child = context.actorOf(Props.empty, "child")
  context.watch(child) // <-- this is the only call needed for registration
  var lastSender = system.deadLetters

  def receive = {
    case "kill" =>
      context.stop(child); lastSender = sender()
    case Terminated(`child`) => lastSender ! "finished"
  }
}

要注意Terminated消息的产生与注册和终止行为所发生的顺序无关。特别地，即使在注册时，被观察的actor已经终止了，监视actor仍然会受到一个Terminated消息。

多次注册并不表示会有多个消息产生，也不保证有且只有一个这样的消息被接收到：如果被监控的actor已经生成了消息并且已经进入了队列，在这个消息被处理之前又发生了另一次注册，则会有第二个消息进入队列，因为对一个已经终止的actor的监控注册操作会立刻导致Terminated消息的产生。

可以使用context.unwatch(target)来停止对另一个actor生存状态的监控。即使Terminated已经加入邮箱，该操作仍有效；一旦调用unwatch，则被观察的actor的Terminated消息就都不会再被处理。

  • 启动Hook

actor启动后，它的preStart方法会被立即执行。

override def preStart() {
// registering with other actors
someService ! Register(self)
}
在actor第一次创建时，将调用此方法。在重新启动期间，它被postRestart的默认实现调用，这意味着通过重写该方法，你可以选择是仅仅在初始化该actor时调用一次，还是为每次重新启动都调用。actor构造函数中的初始化代码将在每个actor实例创建的时候被调用，这也发生在每次重启时。

  • 重启Hook

所有的actor都是被监管的，即与另一个使用某种失败处理策略的actor绑定在一起。如果在处理一个消息的时候抛出了异常，Actor将被重启（详见监管与监控）。这个重启过程包括上面提到的Hook：

要被重启的actor被通知是通过调用preRestart，包含着导致重启的异常以及触发异常的消息；如果重启并不是因为消息处理而发生的，则所携带的消息为None，例如，当一个监管者没有处理某个异常继而被其监管者重启时，或者因其兄弟节点的失败导致的重启。如果消息可用，则消息的发送者通常也可用（即通过调用sender）。

这个方法是用来完成清理、准备移交给新actor实例等操作的最佳位置。其缺省实现是终止所有子actor并调用postStop。

最初调用actorOf的工厂将被用来创建新的实例。
新的actor的postRestart方法被调用时，将携带着导致重启的异常信息。默认实现中，preStart被调用时，就像一个正常的启动一样。
actor的重启只会替换掉原来的actor对象；重启不影响邮箱的内容，所以对消息的处理将在postRestart hook返回后继续。触发异常的消息不会被重新接收。在actor重启过程中，所有发送到该actor的消息将象平常一样被放进邮箱队列中。

警告
要知道失败通知与用户消息的相关顺序不是决定性的。尤其是，在失败以前收到的最后一条消息被处理之前，父节点可能已经重启其子节点了。详细信息请参见“讨论：消息顺序”。

  • 终止 Hook

一个Actor终止后，其postStop hook将被调用，它可以用来，例如取消该actor在其它服务中的注册。这个hook保证在该actor的消息队列被禁止后才运行，即之后发给该actor的消息将被重定向到ActorSystem的deadLetters中。

3. 通过Actor Selection定位Actor

如Actor引用, 路径与地址中所述，每个actor都拥有一个唯一的逻辑路径，此路径是由从actor系统的根开始的父子链构成；它还拥有一个物理路径，如果监管链包含有远程监管者，此路径可能会与逻辑路径不同。这些路径用来在系统中查找actor，例如，当收到一个远程消息时查找收件者，但是它们更直接的用处在于：actor可以通过指定绝对或相对路径（逻辑的或物理的）来查找其它的actor，并随结果获取一个ActorSelection：

1
2
3
4
// will look up this absolute path
context.actorSelection("/user/serviceA/aggregator")
// will look up sibling beneath same supervisor
context.actorSelection("../joe")

其中指定的路径被解析为一个java.net.URI，它以/分隔成路径段。如果路径以/开始，表示一个绝对路径，且从根监管者（”/user”的父亲）开始查找；否则是从当前actor开始。如果某一个路径段为..，会找到当前所遍历到的actor的上一级，否则则会向下一级寻找具有该名字的子actor。 必须注意的是actor路径中的..总是表示逻辑结构，即其监管者。

一个actor selection的路径元素中可能包含通配符，从而允许向匹配模式的集合广播该条消息：

1
2
3
4
// will look all children to serviceB with names starting with worker
context.actorSelection("/user/serviceB/worker*")
// will look up all siblings beneath same supervisor
context.actorSelection("../*")

消息可以通过ActorSelection发送，并且在投递每条消息时 ActorSelection的路径都会被查找。如果selection不匹配任何actor，则消息将被丢弃。

要获得ActorSelection的ActorRef，你需要发送一条消息到selection，然后使用答复消息的sender()引用即可。有一个内置的Identify消息，所有actor会理解它并自动返回一个包含ActorRef的ActorIdentity消息。此消息被遍历到的actor特殊处理为，如果一个具体的名称查找失败（即一个不含通配符的路径没有对应的活动actor），则会生成一个否定结果。请注意这并不意味着应答消息有到达保证，它仍然是一个普通的消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import akka.actor.{ Actor, Props, Identify, ActorIdentity, Terminated }

class Follower extends Actor {
  val identifyId = 1
  context.actorSelection("/user/another") ! Identify(identifyId)

  def receive = {
    case ActorIdentity(`identifyId`, Some(ref)) =>
      context.watch(ref)
      context.become(active(ref))
    case ActorIdentity(`identifyId`, None) => context.stop(self)

  }

  def active(another: ActorRef): Actor.Receive = {
    case Terminated(`another`) => context.stop(self)
  }
}

你也可以通过ActorSelection的resolveOne方法获取ActorSelection的一个ActorRef。如果存在这样的actor，它将返回一个包含匹配的ActorRef的Future。如果没有这样的actor 存在或识别没有在指定的时间内完成，它将以失败告终——akka.actor.ActorNotFound。

如果开启了远程调用，则远程actor地址也可以被查找:

1
context.actorSelection("akka.tcp://app@otherhost:1234/user/serviceB")

4.发送消息

向actor发送消息需使用下列方法之一。

  • !意思是“fire-and-forget”，即异步发送一个消息并立即返回。也称为tell。
  • ?异步发送一条消息并返回一个Future代表一个可能的回应。也称为ask。
    对每一个消息发送者，分别有消息顺序保证。

注意
使用ask有一些性能内涵，因为需要跟踪超时，需要有桥梁将Promise转为ActorRef，并且需要在远程情况下可访问。所以为了性能应该总选择tell，除非只能选择ask。

Tell: Fire-forget

这是发送消息的推荐方式。 不会阻塞地等待消息。它拥有最好的并发性和可扩展性。

1
actorRef ! message

如果是在一个Actor中调用 ，那么发送方的actor引用会被隐式地作为消息的sender(): ActorRef成员一起发送。目的actor可以使用它来向源actor发送回应， 使用sender() ! replyMsg。

如果不是从Actor实例发送的，sender成员缺省为 deadLetters actor引用。

Ask: Send-And-Receive-Future

ask模式既包含actor也包含future，所以它是一种使用模式，而不是ActorRef的方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import akka.pattern.{ ask, pipe }
import system.dispatcher // The ExecutionContext that will be used
case class Result(x: Int, s: String, d: Double)
case object Request

implicit val timeout = Timeout(5 seconds) // needed for `?` below

val f: Future[Result] =
  for {
    x <- ask(actorA, Request).mapTo[Int] // call pattern directly
    s <- (actorB ask Request).mapTo[String] // call by implicit conversion
    d <- (actorC ? Request).mapTo[Double] // call by symbolic name
  } yield Result(x, s, d)

f pipeTo actorD // .. or ..
pipe(f) to actorD

5.接收消息

Actor必须实现receive方法来接收消息：

protected def receive: PartialFunction[Any, Unit]
这个方法应返回一个PartialFunction，例如一个“match/case”子句，消息可以与其中的不同分支进行scala模式匹配。如下例:

1
2
3
4
5
6
7
8
9
10
11
import akka.actor.Actor
import akka.actor.Props
import akka.event.Logging

class MyActor extends Actor {
  val log = Logging(context.system, this)
  def receive = {
    case "test" => log.info("received test")
    case _      => log.info("received unknown message")
  }
}

6.终止Actor

通过调用ActorRefFactory（即ActorContext或ActorSystem）的stop方法来终止一个actor。通常context用来终止子actor，而 system用来终止顶级actor。实际的终止操作是异步执行的，即stop可能在actor被终止之前返回。

actor的终止分两步: 第一步actor将挂起对邮箱的处理，并向所有子actor发送终止命令，然后处理来自子actor的终止消息直到所有的子actor都完成终止，最后终止自己（调用postStop，清空邮箱，向DeathWatch发布Terminated，通知其监管者）。这个过程保证actor系统中的子树以一种有序的方式终止，将终止命令传播到叶子结点并收集它们回送的确认消息给被终止的监管者。如果其中某个actor没有响应（即由于处理消息用了太长时间以至于没有收到终止命令），整个过程将会被阻塞。

在ActorSystem.shutdown()被调用时，系统根监管actor会被终止，以上的过程将保证整个系统的正确终止。

postStop() hook 是在actor被完全终止以后调用的。这是为了清理资源:

1
2
3
override def postStop() {
  // clean up some resources ...
}

注意
由于actor的终止是异步的，你不能马上使用你刚刚终止的子actor的名字；这会导致InvalidActorNameException。你应该 监视watch()正在终止的actor，并在Terminated最终到达后作为回应创建它的替代者。

优雅地终止

如果你需要等待终止过程的结束，或者组合若干actor的终止次序，可以使用gracefulStop：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import akka.pattern.gracefulStop
import scala.concurrent.Await

try {
  val stopped: Future[Boolean] = gracefulStop(actorRef, 5 seconds, Manager.Shutdown)
  Await.result(stopped, 6 seconds)
  // the actor has been stopped
} catch {
  // the actor wasn't stopped within 5 seconds
  case e: akka.pattern.AskTimeoutException =>
}
object Manager {
  case object Shutdown
}

class Manager extends Actor {
  import Manager._
  val worker = context.watch(context.actorOf(Props[Cruncher], "worker"))

  def receive = {
    case "job" => worker ! "crunch"
    case Shutdown =>
      worker ! PoisonPill
      context become shuttingDown
  }

  def shuttingDown: Receive = {
    case "job" => sender() ! "service unavailable, shutting down"
    case Terminated(`worker`) =>
      context stop self
  }
}

当gracefulStop()成功返回时，actor的postStop() hook将会被执行：在postStop()结束和gracefulStop()返回之间存在happens-before边界。

在上面的示例中自定义的Manager.Shutdown消息是发送到目标actor来启动actor的终止过程。你可以使用PoisonPill，但之后在停止目标actor之前，你与其他actor的互动的机会有限。在postStop中，可以处理简单的清理任务。

警告
请记住，actor停止和其名称被注销是彼此异步发生的独立事件。因此，在gracefulStop()返回后。你会发现其名称仍可能在使用中。为了保证正确注销，只在你控制的监管者内，并且只在响应Terminated消息时重用名称，即不是用于顶级actor。

7.Become/Unbecome

升级

Akka支持在运行时对Actor消息循环（即其实现）进行实时替换：在actor中调用context.become方法。become要求一个PartialFunction[Any, Unit]参数作为新的消息处理实现。 被替换的代码被保存在一个栈中，可以被push和pop。

警告
请注意actor被其监管者重启后将恢复其最初的行为。

8.使用PartialFunction链来扩展actor

有时在一些actor中分享共同的行为，或通过若干小的函数构成一个actor的行为是很有用的。这由于actor的receive方法返回一个Actor.Receive（PartialFunction[Any,Unit]的类型别名）而使之成为可能，多个偏函数可以使用PartialFunction#orElse链接在一起。你可以根据需要链接尽可能多的功能，但是你要牢记”第一个匹配”获胜——这在组合可以处理同一类型的消息的功能时会很重要。

例如，假设你有一组actor是生产者Producers或消费者Consumers，然而有时候需要actor分享这两种行为。这可以很容易实现而无需重复代码，通过提取行为的特质和并将actor的receive实现为这些偏函数的组合。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
trait ProducerBehavior {
  this: Actor =>

  val producerBehavior: Receive = {
    case GiveMeThings =>
      sender() ! Give("thing")
  }
}

trait ConsumerBehavior {
  this: Actor with ActorLogging =>

  val consumerBehavior: Receive = {
    case ref: ActorRef =>
      ref ! GiveMeThings

    case Give(thing) =>
      log.info("Got a thing! It's {}", thing)
  }
}

class Producer extends Actor with ProducerBehavior {
  def receive = producerBehavior
}

class Consumer extends Actor with ActorLogging with ConsumerBehavior {
  def receive = consumerBehavior
}

class ProducerConsumer extends Actor with ActorLogging
  with ProducerBehavior with ConsumerBehavior {

  def receive = producerBehavior orElse consumerBehavior
}

// protocol
case object GiveMeThings
case class Give(thing: Any)

不同于继承，相同的模式可以通过组合实现——可以简单地通过委托的偏函数组合成receive方法。